import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.util.*;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IOUtils;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.yarn.webapp.hamlet.Hamlet;


//    java Hw1Grp0 R=/hw1/lineitem.tbl S=/hw1/orders.tbl join:R0=S0 res:S1,R1,R5
//    inputfile1,inputfile2,inputfile1's colume index, inputfile2's column index array,result column array
//        R=src/main/resources/input/lineitem.tbl
//        S=src/main/resources/input/orders.tbl
//        join:R10=S4
//        res:R13,R14,S5,S6

public class LocalJoin {
//    parameters
    public static String inputFileR;
    public static String inputFileS;
    public static int keyIndexR;
    public static int keyIndexS;
    public static ArrayList<Integer> inputFileRColumnIndexArray = new ArrayList<>();
    public static ArrayList<Integer> inputFileSColumnIndexArray = new ArrayList<>();
    public static ArrayList<String> inputFileRColumnName = new ArrayList<>();
    public static ArrayList<String> inputFileSColumnName = new ArrayList<>();
    public static Map<String, ArrayList<ArrayList<String>>> hashJoinMap = new HashMap<String, ArrayList<ArrayList<String>>>();
//  HBase
    public static String tableName = "mytable";
    public static String columnFamily = "res";
    public static HTableDescriptor htd;
    public static HColumnDescriptor cf;
    public static Configuration configuration;
    public static HBaseAdmin hAdmin;
    public static HTable table;

    public static void initInsert() throws IOException {
        htd = new HTableDescriptor(TableName.valueOf(tableName));
        cf = new HColumnDescriptor(columnFamily);
        htd.addFamily(cf);
        configuration = HBaseConfiguration.create();
        hAdmin = new HBaseAdmin(configuration);

        hAdmin.createTable(htd);
        hAdmin.close();
        table = new HTable(configuration, tableName);
        System.out.println("Initialization finished.");
    }

    public static void endInsert() throws IOException {
        table.close();
        System.out.println("Operation finished.");
    }


    //even no order in command,parameters are still parsed
    public static void parseParam(String[] args) {
        for (String str : args) {
            if (str.contains("R=")) {
                inputFileR = str.substring(2);
            }
            else if (str.contains("S=")) {
                inputFileS = str.substring(2);
            }
            else if (str.contains("join:")) {
                int r = str.indexOf("R");
//                TODO:make the calculation of i to a function
                int i;
                for (i=r+1; i<str.length(); ++i) {
                    if (str.charAt(i) < '0' || str.charAt(i) > '9') {
                        break;
                    }
                }
                keyIndexR = Integer.parseInt(str.substring(r+1, i));

                int s = str.indexOf("S");
                for (i=s+1; i<str.length(); ++i) {
                    if (str.charAt(i) < '0' || str.charAt(i) > '9') {
                        break;
                    }
                }
                keyIndexS = Integer.parseInt(str.substring(s+1, i));
            }
            else if (str.contains("res:")) {
                int r = str.indexOf('R');
                while (r != -1) {
                    int i;
                    for (i=r+1; i<str.length(); ++i) {
                        if (str.charAt(i) < '0' || str.charAt(i) > '9') {
                            break;
                        }
                    }
                    inputFileRColumnIndexArray.add(Integer.parseInt(str.substring(r+1, i)));
                    r = str.indexOf('R', r+1);
                }
                int s = str.indexOf("S");
                while (s != -1) {
                    int i;
                    for (i=s+1; i<str.length(); ++i) {
                        if (str.charAt(i) < '0' || str.charAt(i) > '9') {
                            break;
                        }
                    }
                    inputFileSColumnIndexArray.add(Integer.parseInt(str.substring(s+1, i)));
                    s = str.indexOf("S", s+1);
                }
            }
            else {
                System.out.println("Uncomplete command!");
            }
        }
        for (int i : inputFileRColumnIndexArray) {
            inputFileRColumnName.add("R" + i);
        }
        for (int i : inputFileSColumnIndexArray) {
            inputFileSColumnName.add("S" + i);
        }

    }

//    public static BufferedReader readFile(String path) throws IOException, URISyntaxException {
//            String file = "hdfs://localhost:9000";
//            file += path;
//            Configuration conf = new Configuration();
//            FileSystem fs = FileSystem.get(URI.create(file), conf);
//            FSDataInputStream in_stream = fs.open(new Path(file));
//
//            BufferedReader in = new BufferedReader(new InputStreamReader(in_stream));
//            return in;
//    }
    public static BufferedReader readFile(String path) throws IOException, URISyntaxException {
        return new BufferedReader(new FileReader(path));
    }

    public static void showFile(String file) throws IOException, URISyntaxException {
        BufferedReader in = readFile(file);
        String str;
        while ((str = in.readLine()) != null) {
            System.out.println(str);
        }
    }

//assume read from R
    public static Map<String, ArrayList<ArrayList<String>>> formHash(String file) throws IOException, URISyntaxException {
        BufferedReader in = readFile(file);
        String str;
        while ((str = in.readLine()) != null) {
            String[] cols = str.split("\\|");
            String key = cols[keyIndexR];
            ArrayList<String> row = new ArrayList<>();
            for (int i : inputFileRColumnIndexArray) {
                row.add(cols[i]);
            }
//            if exists,add,or new a and add
            if (hashJoinMap.containsKey(key)) {
                hashJoinMap.get(key).add(row);
            } else {
                ArrayList<ArrayList<String>> tmp = new ArrayList<>();
//                add a count for repeat key rows
                ArrayList<String> repArrayList = new ArrayList<>();
                repArrayList.add("0");
                tmp.add(repArrayList);

                hashJoinMap.put(key, tmp);
                hashJoinMap.get(key).add(row);
            }
        }
        return hashJoinMap;
    }

    public static void test(String s1, String s2, String s3) {
        System.out.println(s1);
        System.out.println(s2);
        System.out.println(s3);
    }

    public static void matchJoinMap(String file) throws IOException, URISyntaxException {
        BufferedReader in = readFile(file);
        String str;
        while ((str = in.readLine()) != null) {
            String[] cols = str.split("\\|");
            String key = cols[keyIndexS];

            if (hashJoinMap.containsKey(key)) {
                ArrayList<ArrayList<String>> value = hashJoinMap.get(key);

                String repStr = "";
                int repTime;
//i is ArrayList<ArrayList<String>> index, means row's index;
                for (int i=1; i<value.size(); ++i) {
                    repStr = "";
                    repTime = Integer.parseInt(value.get(0).get(0));
                    if (repTime != 0) {
                        repStr += ".";
                        repStr += String.valueOf(repTime);
                    }
//                    insert R's column
//                    j is index inside an ArrayList<String>, means columns in a row; j can also be used to locate Ri and column value
                    for (int j=0; j<value.get(i).size(); ++j) {
                        String rProjection = value.get(i).get(j);
                        String columnName = inputFileRColumnName.get(j);
                        columnName += repStr;
//                        Put p = new Put(Bytes.toBytes(key));
//                        p.add(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName), Bytes.toBytes(rProjection));
//                        table.put(p);
                    }
//                  insert S's column
                    for (int j=0; j<inputFileSColumnIndexArray.size(); ++j) {
                        String sProjection = cols[inputFileSColumnIndexArray.get(j)];
                        String columnName = inputFileSColumnName.get(j);
                        columnName += repStr;

//                        Put p = new Put(Bytes.toBytes(key));
//                        p.add(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName), Bytes.toBytes(sProjection));
//                        table.put(p);
                    }
//                repeat time add to 1
                repTime++;
                hashJoinMap.get(key).get(0).add(0, String.valueOf(repTime));
                }
            }
        }
    }




//    java Hw1Grp0 R=/hw1/lineitem.tbl S=/hw1/orders.tbl join:R0=S0 res:S1,R1,R5
//    inputfile1,inputfile2,inputfile1's colume index, inputfile2's column index array,result column array

    public static void main(String[] args) throws IOException, URISyntaxException {
        parseParam(args);
//        initInsert();
        formHash(inputFileR);
        matchJoinMap(inputFileS);
//        endInsert();
    }
}